package cn.tedu.micro.demo.delay;

import cn.tedu.micro.demo.client.RabbitChannelManager;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
@Slf4j
public class DelayDemo {
    /**
     * 参考约定名称和绑定关系 声明一套死信的组件结构
     */
    private static final String DELAY_EX="delay_ex";
    private static final String DELAY_Q="delay_q";
    private static final String DLX_EX="dlx_ex";
    private static final String DLX_Q="dlx_q";
    private static final String DLX_RK="dlx_rk";
    @Test
    public void delayDelare() throws IOException {
        //1.拿到channel
        Channel channel = RabbitChannelManager.newChannel();
        //2.声明业务交换机
        channel.exchangeDeclare(DELAY_EX, BuiltinExchangeType.FANOUT);
        //3.死信交换机
        channel.exchangeDeclare(DLX_EX,BuiltinExchangeType.DIRECT);
        //4.业务队列
        Map<String,Object> args=new HashMap<>();
        //绑定死信交换机
        args.put("x-dead-letter-exchange",DLX_EX);
        //死信携带路由key
        args.put("x-dead-letter-routing-key",DLX_RK);
        channel.queueDeclare(DELAY_Q,false,false,false,args);
        //5.死信队列
        channel.queueDeclare(DLX_Q,false,false,false,null);
        //6.业务结构绑定
        channel.queueBind(DELAY_Q,DELAY_EX,"");
        //7.死信结构的绑定
        channel.queueBind(DLX_Q,DLX_EX,DLX_RK);
    }
    @Test
    public void sendDelay() throws IOException{
        //1.客户端操作rabbitmq做什么 都需要channel连接支持
        Channel channel = RabbitChannelManager.newChannel();
        //2.组织消息包含的数据内容
        //2.1 body 消息体
        byte[] body="hello rabbitmq,i am delay message".getBytes(StandardCharsets.UTF_8);
        //2.2 消息属性
        AMQP.BasicProperties.Builder builder=new AMQP.BasicProperties.Builder();
        //固定属性
        builder.expiration("10000");//消息存储超时毫秒数
        AMQP.BasicProperties props = builder.build();
        channel.basicPublish(DELAY_EX,"",props,body);
        log.info("生产发送消息完成");
    }
    //延迟消费逻辑 监听死信队列获取消息
    //延迟消费逻辑 监听死信队列获取消息
    @Test
    public void delayConsume() throws IOException, InterruptedException {
        //1.channel获取
        Channel channel = RabbitChannelManager.newChannel();
        //2.监听消费
        channel.basicConsume(DLX_Q,true,
                (consumerTag,message)->{
                    log.info("消费者接收到消息:{}",new String(message.getBody(),StandardCharsets.UTF_8));
                },
                consumerTag -> {

                }
        );
        Thread.sleep(10000000);
    }

}
